package internal

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/pkg/errors"
	"github.com/zeast/logs"
)

var addressPrefix string = "lark.address."

type EtcdCliConf struct {
	Endpoints      []string
	DelSerCallback func(string, string)
	PutSerCallback func(string, string)
	SetSerCallback func(string, []string) error
}

type EtcdCli struct {
	conf       EtcdCliConf
	cli        *clientv3.Client
	etcdSyncer *EtcdSyncer
	syncBackup bool
}

func NewEtcdCli(conf EtcdCliConf, etcdSyncer *EtcdSyncer) (*EtcdCli, error) {
	ec := &EtcdCli{
		conf:       conf,
		etcdSyncer: etcdSyncer,
	}
	if err := ec.dialEtcd(); err != nil {
		return nil, err
	}

	ec.setServiceMap()

	return ec, nil
}

func (ec *EtcdCli) Watch() {
	for {
		ec.cliWatch()

		ec.dialEtcdForever()
	}
}

func (ec *EtcdCli) cliWatch() {
	wCh := ec.cli.Watch(context.Background(), addressPrefix, clientv3.WithPrefix())

	for {
		select {
		case r := <-wCh:
			if r.Err() != nil {
				logs.Errorf("watch service prefix error:%s", r.Err())
				return
			}

			if len(r.Events) == 0 {
				logs.Errorf("watch service prefix, events is empty")
				return
			}

			for _, ev := range r.Events {
				logs.Debugf("etcd event type:%s, key:%s", ev.Type, ev.Kv.Key)

				name, addr, err := etcdvalue2ServiceAddress(string(ev.Kv.Key))
				if err != nil {
					logs.Errorf("watch service prefix, kv:%s, error:%s", string(ev.Kv.Key), err)
					return
				}

				if ev.Type == mvccpb.DELETE {
					ec.conf.DelSerCallback(name, addr)
					if ec.syncBackup && ec.etcdSyncer != nil {
						ec.etcdSyncer.syncDel(ev.Kv)
					}
				} else if ev.Type == mvccpb.PUT {
					ec.conf.PutSerCallback(name, addr)
					if ec.syncBackup && ec.etcdSyncer != nil {
						ec.etcdSyncer.syncPut(ev.Kv)
					}
				} else {
					logs.Errorf("watch service, unknown events type:%d, key:%s ", ev.Type, ev.Kv.Key)
					return
				}
			}

		}
	}
}

func (ec *EtcdCli) dialEtcdForever() {
	for {
		if err := ec.dialEtcd(); err != nil {
			logs.Error(err)
		} else {
			return
		}

		time.Sleep(time.Second)
	}
}

func (ec *EtcdCli) dialEtcd() error {
	if ec.cli != nil {
		ec.cli.Close()
		ec.cli = nil
	}

	cli, err := clientv3.New(clientv3.Config{
		Endpoints:            ec.conf.Endpoints,
		AutoSyncInterval:     time.Second,
		DialTimeout:          time.Second,
		DialKeepAliveTime:    time.Second,
		DialKeepAliveTimeout: time.Second,
	})

	if err != nil {
		return errors.WithMessage(err, "dial etcd error")
	} else {
		ec.cli = cli
		return nil
	}
}

func (ec *EtcdCli) setServiceMap() error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()

	resp, err := ec.cli.Get(ctx, addressPrefix, clientv3.WithPrefix())
	if err != nil {
		logs.Errorf("init service map. error:%s", err)
		return err
	}

	var mm = make(map[string][]string)
	for _, kv := range resp.Kvs {
		name, addr, err := etcdvalue2ServiceAddress(string(kv.Key))
		if err != nil {
			logs.Errorf("init service map error. key:%s, error:%s", string(kv.Key), err)
			return err
		}
		mm[name] = append(mm[name], addr)
	}

	for n, m := range mm {
		if err := ec.conf.SetSerCallback(n, m); err != nil {
			return err
		} else {
			logs.Debugf(" init service success, name:%s, data:%v", n, m)
		}
	}

	return nil
}

//返回服务名，服务地址
func etcdvalue2ServiceAddress(s string) (string, string, error) {
	data := strings.SplitN(s, ".", 5)
	if len(data) != 5 {
		return "", "", fmt.Errorf("etcd key is wrong. %s", s)
	}

	return data[2], data[4], nil
}

func (ec *EtcdCli) WatchUseBackup(value interface{}) {
	logs.Infof("watch use backup %v", value)
	if useBackup, ok := value.(bool); ok {
		ec.syncBackup = !useBackup
		if ec.syncBackup {
			if ec.etcdSyncer != nil {
				ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
				defer cancel()

				resp, err := ec.cli.Get(ctx, addressPrefix, clientv3.WithPrefix())
				if err != nil {
					logs.Errorf("init service map. error:%s", err)
					return
				}
				ec.etcdSyncer.syncInit(resp.Kvs)
			}
		}
	}
}
